package com.nbcio.iot.ingest;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.nbcio.iot.domain.entity.IotDataPoint;
import com.nbcio.iot.domain.entity.IotDataRecord;
import com.nbcio.iot.domain.entity.IotDevice;
import com.nbcio.iot.mapper.IotDataPointMapper;
import com.nbcio.iot.mapper.IotDataRecordMapper;
import com.nbcio.iot.mapper.IotDeviceMapper;
import com.nbcio.iot.rule.RuleRuntimeService;
import com.nbcio.iot.service.NodeRedService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

import com.nbcio.common.tenant.helper.TenantHelper;
import com.nbcio.common.websocket.websocket.WebSocketServer;

@Slf4j
@Service
@RequiredArgsConstructor
public class IotIngestionService {

    private final ObjectMapper objectMapper = new ObjectMapper();
    private final IotDeviceMapper deviceMapper;
    private final IotDataPointMapper dataPointMapper;
    private final IotDataRecordMapper dataRecordMapper;
    private final RuleRuntimeService ruleRuntime;
    private final NodeRedService nodeRedService;
    private final com.nbcio.iot.mapper.IotProductMapper productMapper;
    private final WebSocketServer webSocketServer;

	// 扩展忽略集合，防止将非点位元数据写入点位
	private static final Set<String> META_KEYS = java.util.Set.of(
		"deviceCode", "deviceId", "productId", "flowId",
		"timestamp", "ts", "time", "messageType", "payload"
	);

    public void handleIncoming(String topic, String payload) {
        try {
            log.info("[INGEST] incoming topic={}, payloadPreview={}", topic, payload == null ? null : (payload.length() > 256 ? payload.substring(0, 256) + "..." : payload));
            // 解析 topic: devices/{deviceCode}/up
            String[] parts = topic.split("/");
            if (parts.length < 3) {
                log.warn("[INGEST] ignore topic: {} (parts < 3)", topic);
                return;
            }
            String deviceCode = parts[1];

            IotDevice device = deviceMapper.selectOne(new LambdaQueryWrapper<IotDevice>()
                .eq(IotDevice::getDeviceCode, deviceCode)
                .last("limit 1"));
            if (device == null) {
                device = new IotDevice();
                device.setDeviceCode(deviceCode);
                device.setDeviceName(deviceCode);
				// 注意：生产环境应由上游分配租户，这里仅兜底避免空
				device.setTenantId("000000");
                deviceMapper.insert(device);
                log.info("[INGEST] auto-created device: {}", deviceCode);
            } else {
                log.info("[INGEST] resolved device id={}, code={}", device.getDeviceId(), device.getDeviceCode());
            }

			final IotDevice finalDevice = device;
			TenantHelper.dynamic(finalDevice.getTenantId(), () -> {
				try {
            JsonNode root = objectMapper.readTree(payload);

					// 防回环：带有 messageType/flowId 的内部业务消息直接忽略（例如规则执行转发的消息）
					if ((root.has("messageType") && !root.get("messageType").isNull()) || (root.has("flowId") && !root.get("flowId").isNull())) {
						log.info("[INGEST] skip internal message (has messageType/flowId), topic={}, device={}", topic, deviceCode);
						return;
					}

            log.info("[INGEST] payload root keys: {}", root.isObject() ? root.fieldNames().hasNext() ? java.util.Arrays.toString(java.util.stream.StreamSupport.stream(((Iterable<String>)() -> root.fieldNames()).spliterator(), false).toArray(String[]::new)) : "{}" : root.getNodeType());

            int saved = 0;

            // 优先展开 metrics
            if (root.has("metrics") && root.get("metrics").isObject()) {
                JsonNode metrics = root.get("metrics");
                Iterator<Map.Entry<String, JsonNode>> metricFields = metrics.fields();
                while (metricFields.hasNext()) {
                    Map.Entry<String, JsonNode> entry = metricFields.next();
                    String pointCode = entry.getKey();
                    JsonNode valueNode = entry.getValue();
							saved += saveOnePoint(finalDevice, pointCode, valueNode) ? 1 : 0;
                }
            }

            // 同时处理根层非元数据、非metrics字段
            Iterator<Map.Entry<String, JsonNode>> fields = root.fields();
            while (fields.hasNext()) {
                Map.Entry<String, JsonNode> entry = fields.next();
                String pointCode = entry.getKey();
                if (META_KEYS.contains(pointCode) || "metrics".equals(pointCode)) {
                    continue;
                }
                JsonNode valueNode = entry.getValue();
						saved += saveOnePoint(finalDevice, pointCode, valueNode) ? 1 : 0;
            }

            if (saved > 0) {
                log.info("[INGEST] saved {} record(s) from topic={}, device={}", saved, topic, deviceCode);
                // 标记设备在线并更新最近在线时间
                try {
                    finalDevice.setDeviceStatus("1"); // 1=在线
                    finalDevice.setLastOnlineTime(java.time.LocalDateTime.now());
                    deviceMapper.updateById(finalDevice);
                } catch (Exception e) {
                    log.warn("[INGEST] update device online status failed: {}", e.getMessage());
                }

                // 通过 WebSocket 广播设备在线事件
                try {
                    String wsMsg = String.format("{\"type\":\"device_online\",\"deviceCode\":\"%s\",\"timestamp\":%d}",
                            finalDevice.getDeviceCode(), System.currentTimeMillis());
                    webSocketServer.pushMessage(wsMsg);
                } catch (Exception e) {
                    log.warn("[INGEST] websocket broadcast failed: {}", e.getMessage());
                }
                try {
                    java.util.Map<String,Object> msg = new java.util.HashMap<>();
							msg.put("tenantId", finalDevice.getTenantId());
							msg.put("deviceId", finalDevice.getDeviceId());
							msg.put("deviceCode", finalDevice.getDeviceCode());
                    msg.put("payload", root);
                    msg.put("ts", System.currentTimeMillis());

                    // 规则解析：设备 > 产品 > 默认
                    java.util.Map<String,Object> flowCfg = null;
							if (finalDevice.getFlowId() != null) {
								flowCfg = nodeRedService.getFlowConfigById(finalDevice.getFlowId());
                    }
							if ((flowCfg == null || flowCfg.isEmpty()) && finalDevice.getProductId() != null) {
								com.nbcio.iot.domain.entity.IotProduct product = productMapper.selectById(finalDevice.getProductId());
                        if (product != null && product.getFlowId() != null) {
                            flowCfg = nodeRedService.getFlowConfigById(product.getFlowId());
                        }
                    }
                    if (flowCfg == null || flowCfg.isEmpty()) {
                        flowCfg = nodeRedService.getDefaultActiveFlowConfig();
                    }

                    if (flowCfg != null && !flowCfg.isEmpty()) {
                        log.info("[INGEST] invoking ruleRuntime.process for device={} with flow present", deviceCode);
                        ruleRuntime.process(msg, flowCfg);
                    } else {
                        log.info("[INGEST] no rule flow resolved for device={}, skip ruleRuntime", deviceCode);
                    }
                } catch (Exception ex) {
                    log.warn("[INGEST] ruleRuntime.process error: {}", ex.getMessage());
                }
            } else {
                log.info("[INGEST] no fields saved from payload for topic={}, device={}", topic, deviceCode);
            }
        } catch (Exception e) {
            log.error("ingest error", e);
        }
			});
		} catch (Exception e) {
			log.error("ingest error", e);
		}
	}

	/**
	 * 直接持久化采集到的指标（在调用处已确保租户上下文）。
	 */
	public int saveMetrics(IotDevice device, Map<String, Object> metrics) {
		if (device == null || metrics == null || metrics.isEmpty()) {
			return 0;
		}
		int saved = 0;
		for (Map.Entry<String, Object> entry : metrics.entrySet()) {
			String pointCode = entry.getKey();
			if (META_KEYS.contains(pointCode)) {
				continue;
			}
			JsonNode valueNode = objectMapper.valueToTree(entry.getValue());
			saved += saveOnePoint(device, pointCode, valueNode) ? 1 : 0;
		}
        // 若有数据入库，则标记设备在线并广播
        if (saved > 0) {
            try {
                device.setDeviceStatus("1"); // 1=在线
                device.setLastOnlineTime(java.time.LocalDateTime.now());
                deviceMapper.updateById(device);
            } catch (Exception e) {
                log.warn("[INGEST] update device online status (saveMetrics) failed: {}", e.getMessage());
            }
            try {
                String wsMsg = String.format("{\"type\":\"device_online\",\"deviceCode\":\"%s\",\"timestamp\":%d}",
                        device.getDeviceCode(), System.currentTimeMillis());
                webSocketServer.pushMessage(wsMsg);
            } catch (Exception e) {
                log.warn("[INGEST] websocket broadcast (saveMetrics) failed: {}", e.getMessage());
            }
        }
        return saved;
    }

    private boolean saveOnePoint(IotDevice device, String pointCode, JsonNode valueNode) {
        IotDataPoint point = dataPointMapper.selectOne(new LambdaQueryWrapper<IotDataPoint>()
            .eq(IotDataPoint::getDeviceId, device.getDeviceId())
            .eq(IotDataPoint::getPointCode, pointCode)
            .last("limit 1"));
        if (point == null) {
            point = new IotDataPoint();
            point.setDeviceId(device.getDeviceId());
            point.setPointCode(pointCode);
            point.setPointName(pointCode);
            point.setDataType(valueNode.isNumber() ? "number" : (valueNode.isTextual() ? "string" : "object"));
            dataPointMapper.insert(point);
            log.info("[INGEST] created point deviceId={}, pointCode={}", device.getDeviceId(), pointCode);
        }
        IotDataRecord record = new IotDataRecord();
        record.setDeviceId(device.getDeviceId());
        record.setPointId(point.getPointId());
        
        // 处理数据值，避免VARCHAR(500)字段长度超限
        String dataValue;
        if (valueNode.isContainerNode()) {
            String jsonStr = valueNode.toString();
			// 限制JSON字符串长度为500字符，考虑截断标记的长度
            if (jsonStr.length() > 500) {
				int maxLength = 500 - 15; // 为 "[TRUNCATED]" 预留空间
				dataValue = jsonStr.substring(0, maxLength) + "...[TRUNCATED]";
                log.warn("[INGEST] JSON数据值过长已截断，设备ID: {}, 点位: {}, 原始长度: {}, 截断后长度: {}", 
                    device.getDeviceId(), pointCode, jsonStr.length(), dataValue.length());
            } else {
                dataValue = jsonStr;
            }
        } else {
            dataValue = valueNode.asText();
			// 对文本值也进行长度限制，考虑截断标记的长度
            if (dataValue.length() > 500) {
				int maxLength = 500 - 15; // 为 "[TRUNCATED]" 预留空间
				dataValue = dataValue.substring(0, maxLength) + "...[TRUNCATED]";
                log.warn("[INGEST] 文本数据值过长已截断，设备ID: {}, 点位: {}, 原始长度: {}, 截断后长度: {}", 
                    device.getDeviceId(), pointCode, dataValue.length(), dataValue.length());
            }
        }
        
        record.setDataValue(dataValue);
        record.setDataTime(new Date());
        record.setQuality("GOOD");
        dataRecordMapper.insert(record);
        return true;
    }
} 